package org.apm.data.pusher;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.ClearScrollRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;

import com.alibaba.fastjson.JSONObject;

public class UpdateSource {
    private static Logger LOG = LogManager.getLogger(UpdateSource.class);
    private static final ScheduledExecutorService INDEXSERVICE = Executors.newScheduledThreadPool(8);

    private static final int SCROLL_TIME = 60000;
    private static final int SCROLL_SIZE = 1000;
    
    private static void copy(final String index, final Client src, final BulkProcessor srcp) {
        QueryBuilder query = QueryBuilders.boolQuery()
                .filter(QueryBuilders.termsQuery("channelType", "直接访问","direct","search","internal","external","外来连接"));
        
        SearchResponse scrollResp = src.prepareSearch(index).setQuery(query).setScroll(new TimeValue(SCROLL_TIME))
                .setSize(SCROLL_SIZE).execute().actionGet();

        ClearScrollRequestBuilder clearRequest = src.prepareClearScroll();
        while (true) {
            for (SearchHit hit : scrollResp.getHits()) {
                JSONObject json = JSONObject.parseObject(hit.getSourceAsString());
                String channelType = json.getString("channelType");
                switch (channelType) {
                    case "direct":
                    case "直接访问":
                        json.put("channelType", "直接访问");
                        json.put("source", "直接访问");
                        json.put("supplement", "直接访问");
                        break;
                    case "search":
                        json.put("channelType", "搜索引擎");
                        break;
                    case "internal":
                        json.put("channelType", "内部来源");
                        break;
                    case "external":
                    case "外来连接":
                        json.put("channelType", "外部链接");
                        break;
                    default:
                        continue;
                }
                
                IndexRequest indexRequest = new IndexRequest(index, hit.getType(), hit.getId())
                        .source(json.toJSONString());
                System.out.println(json.toJSONString());
                srcp.add(indexRequest);
            }

            scrollResp = src.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(SCROLL_TIME))
                    .execute().actionGet();

            // Break condition: No hits are returned
            if (scrollResp.getHits().getHits().length == 0) {
                break;
            } else {
                LOG.info("[Analyzer-After][" + index +"]" + scrollResp.getHits().getHits().length + ".");
            }
        }

        // 清缓存
        clearRequest.addScrollId(scrollResp.getScrollId()).execute().actionGet().isSucceeded();
        clearRequest.execute().actionGet();
    }
    
    public static void main(String[] args) throws InterruptedException {
        //copy("apm2.0-172.10.105.112-pagemonitor-1" , CommonMethod.onlineClient(), CommonMethod.EsBulkProcessor.ONLINE.processor());
        copy("apm2.0-172.10.105.112-channel-1" , CommonMethod.devClient(), CommonMethod.EsBulkProcessor.DEV.processor());
        
        Thread.sleep(5000);
    }
}
